package cn.com.shaom.learning.sb.batch.config;

import cn.com.shaom.learning.sb.batch.Person;
import cn.com.shaom.learning.sb.batch.service.CvsItemProcessor;
import cn.com.shaom.learning.sb.batch.service.CvsItemValidator;
import cn.com.shaom.learning.sb.batch.service.CvsJobListener;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

/**
 * ClassName: CvsBatchConfig
 * Description:
 * Date: 2017/12/28 15:49 【需求编号】
 *
 * @author Shaom
 * @version V1.0.0
 */
@Configuration
@EnableBatchProcessing
public class CvsBatchConfig {


    /**
     * ItemReader，读配置
     *
     * @return
     * @StepScope 实现参数前置
     */
    @Bean
    public ItemReader<Person> reader() {
        //使用 FlatFileItemReader 读取文件
        FlatFileItemReader<Person> reader = new FlatFileItemReader<>();

        // 设置CVS 文件的路径
        reader.setResource(new ClassPathResource("people.cvs"));

        // 对cvs 文件的数据和领域模型类做对应映射
        reader.setLineMapper(new DefaultLineMapper<Person>() {{
            setLineTokenizer(new DelimitedLineTokenizer() {{
                setNames(new String[]{"name", "age", "nation", "address"});
            }});

            setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                setTargetType(Person.class);
            }});

        }});
        return reader;
    }

    /**
     * ItemProcessor，业务处理定义
     *
     * @return
     */
    @Bean
    public ItemProcessor<Person, Person> processor() {
        //使用我们自定义的数据处理
        CvsItemProcessor processor = new CvsItemProcessor();
        //指定校验器CvsItemValidator
        processor.setValidator(cvsItemValidator());
        return processor;
    }

    /**
     * 定义校验的bean
     *
     * @return
     */
    @Bean
    public CvsItemValidator cvsItemValidator() {
        return new CvsItemValidator();
    }

    /**
     * ItemWriter定义,写
     *
     * @param dataSource
     * @return
     */
    @Bean
    public ItemWriter<Person> writer(DataSource dataSource) {//Spring能让容器中已有的 Bean 以参数的形式注入

        // 使用JDBC批处理 JdbcBatchItemWriter 来写数据到数据库
        JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());

        String sql = "insert into person (name, age, nation, address)" + "values (:name, :age, :nation, :address)";
        //设置批处理的SQL
        writer.setSql(sql);
        writer.setDataSource(dataSource);
        return writer;
    }

    /**
     * JobRepository定义
     *
     * @param dataSource
     * @param transactionManager
     * @return
     * @throws Exception
     */
    @Bean
    public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
        JobRepositoryFactoryBean jobRepository = new JobRepositoryFactoryBean();
        jobRepository.setDataSource(dataSource);
        jobRepository.setTransactionManager(transactionManager);
        jobRepository.setDatabaseType("mysql");
        return jobRepository.getObject();
    }

    /**
     * JobLauncher 定义
     * 作业调度器
     *
     * @param dataSource
     * @param transactionManager
     * @return
     * @throws Exception
     */
    @Bean
    public SimpleJobLauncher jobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository(dataSource, transactionManager));
        return jobLauncher;
    }

    /**
     * Job定义
     *
     * @param jobs
     * @param s1
     * @return
     */
    @Bean
    public Job importJob(JobBuilderFactory jobs, Step s1) {
        return jobs.get("job")
                .incrementer(new RunIdIncrementer())
                //为Job 指定Step
                .flow(s1)
                .end()
                //绑定监听
                .listener(cvsJobListener())
                .build();
    }

    /**
     * Job 监听
     *
     * @return
     */
    @Bean
    public CvsJobListener cvsJobListener() {
        return new CvsJobListener();
    }


    /**
     * Step 定义
     *
     * @param stepBuilderFactory
     * @param reader
     * @param writer
     * @param processor
     * @return
     */
    @Bean
    public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<Person> reader, ItemWriter<Person> writer,
                      ItemProcessor<Person, Person> processor) {

        return stepBuilderFactory
                .get("step1")
                //批处理每次提交65000条数据
                .<Person, Person>chunk(65000)
                //给Step 绑定 reader
                .reader(reader)
                //给Step 绑定 processor
                .processor(processor)
                //给Step 绑定 writer
                .writer(writer)
                .build();
    }

}
